package com.shujia.gma.dim

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo1MysqlRegionDIm {
  /**
    *
    * 全量数据导入
    *
    *
    * 在上线的时候先进行一次全量数据导入，在启动增量数据导入
    *
    *
    */
  def main(args: Array[String]): Unit = {

    val fsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //计划器
      .inStreamingMode()
      .build()

    val fsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    fsEnv.setParallelism(1)
    val fsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(fsEnv, fsSettings)


    fsTableEnv.executeSql(
      """
        |CREATE TABLE mysql_base_province (
        |  id int,
        |  name STRING,
        |  region_id int,
        |  area_code STRING
        |) WITH (
        | 'connector' = 'jdbc',
        | 'table-name' = 'base_province',
        | 'url' = 'jdbc:mysql://master:3306/gma',
        | 'username' = 'root',
        | 'password' ='123456'
        |)
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |CREATE TABLE mysql_base_region (
        |  id int,
        |  region_name STRING
        |) WITH (
        | 'connector' = 'jdbc',
        | 'table-name' = 'base_region',
        | 'url' = 'jdbc:mysql://master:3306/gma',
        | 'username' = 'root',
        | 'password' ='123456'
        |)
        |
      """.stripMargin)

    /**
      * 在hbase中先创建维表
      * create_namespace 'dim'
      * create 'dim:base_region' ,'i'
      *
      */

    fsTableEnv.executeSql(
      """
        |CREATE TABLE dim_base_region (
        | id int,
        | i ROW<name STRING,region_name STRING,area_code STRING>,
        | PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'dim:base_region',
        | 'zookeeper.quorum' = 'master:2181,node1,2181,node2:2181'
        |)
        |
      """.stripMargin)


    /**
      * 执行关联将维表数据保存到hbase
      *
      */

   /* fsTableEnv.executeSql(
      """
        |
        |insert into dim_base_region
        |select a.id, ROW(name,region_name,area_code) as i from
        |mysql_base_province as a
        |join mysql_base_region  as b
        |on a.region_id=b.id
        |
      """.stripMargin)*/

    fsTableEnv.sqlQuery(
      """
        |
        |select * from dim_base_region
        |
          """.stripMargin).toAppendStream[Row].print()

    fsEnv.execute()


  }
}
